package cn.monkey.transport.netty.duplex;

import cn.monkey.transport.core.Inbound;
import cn.monkey.transport.netty.NettyOutbound;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;

import java.util.List;

public class SimpleDuplexOutbound implements DuplexOutbound {
    protected final NettyOutbound nettyOutbound;
    protected final Inbound duplexInbound;
    protected final Disposable disposable;
    protected final Scheduler scheduler;

    public SimpleDuplexOutbound(NettyOutbound nettyOutbound,
                                Inbound duplexInbound,
                                Scheduler scheduler) {
        this.nettyOutbound = nettyOutbound;
        this.duplexInbound = duplexInbound;
        this.scheduler = scheduler;
        this.disposable = this.init();
    }

    private Disposable init() {
        return this.inbound()
                .receive(Object.class)
                .subscribeOn(this.scheduler)
                .subscribe(this.nettyOutbound()::writeAndFlush);
    }

    @Override
    public NettyOutbound nettyOutbound() {
        return this.nettyOutbound;
    }

    @Override
    public Inbound inbound() {
        return this.duplexInbound;
    }

    @Override
    public <T> Flux<T> receive(Class<T> clazz) {
        return this.inbound().receive(clazz);
    }

    @Override
    public Mono<Void> close() {
        this.disposable.dispose();
        List<Mono<Void>> list = List.of(this.nettyOutbound().close(), this.inbound().close());
        return Flux.merge(list).then();
    }
}
